package edu.wisc.hevea.messaging;

import java.util.concurrent.Executors;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;

import com.google.protobuf.Service;
import com.googlecode.protobuf.pro.duplex.CleanShutdownHandler;
import com.googlecode.protobuf.pro.duplex.PeerInfo;
import com.googlecode.protobuf.pro.duplex.RpcClientChannel;
import com.googlecode.protobuf.pro.duplex.RpcConnectionEventNotifier;

import com.googlecode.protobuf.pro.duplex.execute.ThreadPoolCallExecutor;
import com.googlecode.protobuf.pro.duplex.listener.RpcConnectionEventListener;
import com.googlecode.protobuf.pro.duplex.logging.CategoryPerServiceLogger;
import com.googlecode.protobuf.pro.duplex.server.DuplexTcpServerBootstrap;

import edu.wisc.hevea.broker.MessageBroker;
import edu.wisc.hevea.topic.TopicManager;

public class MessagingServer implements Runnable {

	private static Log log = LogFactory.getLog(MessagingServer.class);
	private String serverHostname;
	private int serverPort;
	private TopicManager topicManager;
	private MessageBroker messageBroker;

	public MessagingServer(MessageBroker messageBroker,TopicManager topicManager) {
		String[] sub = messageBroker.getMyHostName().split(":");
		this.serverHostname = sub[0];
		this.serverPort = Integer.parseInt(sub[1]);
		this.topicManager = topicManager;
		this.messageBroker = messageBroker;
	}

	@Override
	public void run() {

		PeerInfo serverInfo = new PeerInfo(serverHostname, serverPort);

		// RPC payloads are uncompressed when logged - so reduce logging
		CategoryPerServiceLogger logger = new CategoryPerServiceLogger();
		logger.setLogRequestProto(false);
		logger.setLogResponseProto(false);

		// Configure the server.
		DuplexTcpServerBootstrap bootstrap = new DuplexTcpServerBootstrap(
				serverInfo, new NioServerSocketChannelFactory(
						Executors.newCachedThreadPool(),
						Executors.newCachedThreadPool()),
				new ThreadPoolCallExecutor(10, 10), logger);

		CleanShutdownHandler shutdownHandler = new CleanShutdownHandler();
		shutdownHandler.addResource(bootstrap);

		// setup a RPC event listener - it just logs what happens
		RpcConnectionEventNotifier rpcEventNotifier = new RpcConnectionEventNotifier();
		RpcConnectionEventListener listener = new RpcConnectionEventListener() {

			@Override
			public void connectionReestablished(RpcClientChannel clientChannel) {
				log.info("connectionReestablished " + clientChannel);
			}

			@Override
			public void connectionOpened(RpcClientChannel clientChannel) {
				log.info("connectionOpened " + clientChannel);
			}

			@Override
			public void connectionLost(RpcClientChannel clientChannel) {
				log.info("connectionLost " + clientChannel);
			}

			@Override
			public void connectionChanged(RpcClientChannel clientChannel) {
				log.info("connectionChanged " + clientChannel);
			}
		};
		rpcEventNotifier.setEventListener(listener);
		bootstrap.registerConnectionEventListener(rpcEventNotifier);

		Service messagingService = Messaging.MessagingService
				.newReflectiveService(new MessagingServiceImpl(topicManager,
						messageBroker));
		bootstrap.getRpcServiceRegistry().registerService(messagingService);

		// Bind and start to accept incoming connections.
		bootstrap.bind();
		log.info("Serving " + bootstrap);
	}

}
